package com.mall.hadoop.job.tfirstgardenstatistics;

import java.io.IOException;
import java.util.Map;

import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import com.mall.untils.DateUtils;

import net.sf.json.JSONObject;
/**
 * 全国圆所统计省、市、区
 * @author liugb
 *
 */
public class FirstLoginTimeMap extends Mapper<LongWritable, Text, Text, IntWritable>{

	@Override
	protected void map(LongWritable key, Text value, Context context)throws IOException, InterruptedException {
		try {
			Map<String,Object> map = JSONObject.fromObject(value.toString());
			String p = MapUtils.getString(map, "province");
			String c = MapUtils.getString(map, "city");
			String a = MapUtils.getString(map, "area");	
			//登陆时间		
			String starttime = MapUtils.getString(map, "starttime");
			String endtime = MapUtils.getString(map, "endtime");
			Long start = DateUtils.getTimestampOfSS(String.valueOf(starttime));
			Long end = DateUtils.getTimestampOfSS(String.valueOf(endtime));
			int counttime = (int) (end-start);		
			if(!StringUtils.isEmpty(p)){				
				context.write(new Text(p+"zh_1"), new IntWritable(counttime));
			}
			if(!StringUtils.isEmpty(c)){
				context.write(new Text(p+c+"xh_1"), new IntWritable(counttime));
			}
			if(!StringUtils.isEmpty(a)){
				context.write(new Text(p+c+a+"mh_1"), new IntWritable(counttime));
			}				
			
		} catch (Exception e) {
			// TODO: handle exception
			e.printStackTrace();
		}
	}

}
